/**
 *  Copyright (c)  2011-2020 Panguso, Inc.
 *  All rights reserved.
 *
 *  This software is the confidential and proprietary information of Panguso, 
 *  Inc. ("Confidential Information"). You shall not
 *  disclose such Confidential Information and shall use it only in
 *  accordance with the terms of the license agreement you entered into with Panguso.
 */

package storage.util;

import java.nio.ByteBuffer;
import java.util.HashMap;
import java.util.List;
import java.util.logging.Level;
import java.util.logging.Logger;
import org.apache.cassandra.thrift.Cassandra;
import org.apache.cassandra.thrift.Column;
import org.apache.cassandra.thrift.ColumnParent;
import org.apache.cassandra.thrift.ColumnPath;
import org.apache.cassandra.thrift.ConsistencyLevel;
import org.apache.cassandra.thrift.IndexClause;
import org.apache.cassandra.thrift.IndexExpression;
import org.apache.cassandra.thrift.IndexOperator;
import org.apache.cassandra.thrift.KeyRange;
import org.apache.cassandra.thrift.KeySlice;
import org.apache.cassandra.thrift.SlicePredicate;
import org.apache.cassandra.thrift.SliceRange;
import org.apache.thrift1.protocol.TBinaryProtocol;
import org.apache.thrift1.protocol.TProtocol;
import org.apache.thrift1.transport.TFramedTransport;
import org.apache.thrift1.transport.TSocket;
import org.apache.thrift1.transport.TTransport;
import org.apache.thrift1.transport.TTransportException;
import cn.edu.thu.log.web.service.PropertiesReaderServices;
import cn.edu.thu.log.web.service.impl.PropertiesReaderServicesImpl;

/**
 * 
 * @author bloodyparadise
 */
public class CassandraUtil {

	private String test;

	public String getTest() {
		return test;
	}

	/**
	 * 
	 * @param test parameter
	 * @return
	 */
	public void setTest(String test) {
		this.test = test;
	}
	
	/**
	 * 
	 * @param test parameter
	 * @return
	 */
	public CassandraUtil(String test) {
		this.test = test;
	}
	private static HashMap<String, TTransport> ttMap = new HashMap<String, TTransport>();
	private static HashMap<String, Cassandra.Client> clientMap = new HashMap<String, Cassandra.Client>();

	/**
	 * 向链接库获得一个指定keyspace的数据库链接
	 * 
	 * @param keySpace
	 *            指定的keyspace名称
	 * @return 返回的数据库连接
	 */
	/**
	 * 
	 * @param keySpace parameter
	 * @return
	 */
	public static Cassandra.Client getInstance(final String keySpace) {
		TTransport tt = ttMap.get(keySpace);
		Cassandra.Client client = clientMap.get(keySpace);
		PropertiesReaderServices x1 = PropertiesReaderServicesImpl.instance();
		if (tt == null) {
			try {
				// System.out.println(x1.getProperties("dbConnection"));
				tt = new TFramedTransport(new TSocket(
						x1.getProperties("dbConnection"), 9160));
				TProtocol proto = new TBinaryProtocol(tt);
				client = new Cassandra.Client(proto);
				if (tt == null) {
					return null;
				}
				tt.open();
				ttMap.put(keySpace, tt);
				clientMap.put(keySpace, client);
			} catch (TTransportException ex) {
				Logger.getLogger(CassandraUtil.class.getName()).log(
						Level.SEVERE, null, ex);
			}
		}

		return client;
	}

	/**
	 * 
	 * @param keySpace parameter
	 * @return
	 */
	public static void open(final String keySpace) {
		TTransport tt = ttMap.get(keySpace);
		if (tt != null) {
			if (!tt.isOpen()) {
				try {
					tt.open();
				} catch (TTransportException ex) {
					Logger.getLogger(CassandraUtil.class.getName()).log(
							Level.SEVERE, null, ex);
				}
			}
		}
	}

	/**
	 * 关闭一个指定keyspace的数据库连接
	 * 
	 * @param keySpace
	 *            指定的keyspace名称
	 */
	/**
	 * 
	 * @param keySpace parameter
	 * @return
	 */
	public static void close(final String keySpace) {
		TTransport tt = ttMap.get(keySpace);
		if (tt != null) {
			if (tt.isOpen()) {
				tt.close();
			}
		}
	}

	/**
	 * 
	 * @param start parameter
	 * @param end parameter
	 * @return
	 */
	private static SlicePredicate buildSlicePredicate(String start, String end) {

		SlicePredicate slicePredicate = new SlicePredicate();
		SliceRange sliceRange = new SliceRange();
		sliceRange.setStart(ByteBuffer.wrap(start.getBytes()));
		sliceRange.setFinish(ByteBuffer.wrap(end.getBytes()));
		slicePredicate.setSlice_range(sliceRange);

		return slicePredicate;
	}

	/**
	 * 
	 * @param start parameter
	 * @param end parameter
	 * @return
	 */
	private static KeyRange buildKeyRange(String start, String end) {
		KeyRange keyrange = new KeyRange();
		keyrange.setStart_key(ByteBuffer.wrap(start.getBytes()));
		keyrange.setEnd_key(ByteBuffer.wrap(end.getBytes()));

		return keyrange;
	}

	/**
	 * 
	 * @param column parameter
	 * @param value parameter
	 * @param op parameter
	 * @param startKey parameter
	 * @return
	 */
	private static IndexClause buildIndexClause(String column, String value,
			IndexOperator op, String startKey) {
		IndexClause indexClause = new IndexClause();
		IndexExpression ie = new IndexExpression();
		ie.setColumn_name(ByteBuffer.wrap(column.getBytes()));
		ie.setOp(op);
		ie.setValue(ByteBuffer.wrap(value.getBytes()));
		indexClause.addToExpressions(ie);
		indexClause.setStart_key(ByteBuffer.wrap(startKey.getBytes()));
		// indexClause.setCount(limitrow);

		return indexClause;
	}

	/**
	 * 
	 * @param column1 parameter
	 * @param value1 parameter
	 * @param column2 parameter
	 * @param value2 parameter
	 * @param op parameter
	 * @param startKey parameter
	 * @return
	 */
	private static IndexClause buildIndexClause(String column1, String value1,
			String column2, String value2, IndexOperator op, String startKey) {
		IndexClause indexClause = new IndexClause();
		IndexExpression ie = new IndexExpression();
		ie.setColumn_name(ByteBuffer.wrap(column1.getBytes()));
		ie.setOp(op);
		ie.setValue(ByteBuffer.wrap(value1.getBytes()));
		indexClause.addToExpressions(ie);
		IndexExpression ie1 = new IndexExpression();
		ie1.setColumn_name(ByteBuffer.wrap(column2.getBytes()));
		ie1.setOp(op);
		ie1.setValue(ByteBuffer.wrap(value2.getBytes()));
		indexClause.addToExpressions(ie1);
		indexClause.setStart_key(ByteBuffer.wrap(startKey.getBytes()));
		return indexClause;
	}

	/**
	 * 
	 * @param column parameter
	 * @param value parameter
	 * @return
	 */
	private static Column buildColumn(String columnName, String value) {
		Column column = new Column();
		column.setName(columnName.getBytes());
		column.setValue(value.getBytes());
		column.setTimestamp(System.currentTimeMillis());

		return column;
	}

	/**
	 * 
	 * @param columnFamily parameter
	 * @return
	 */
	private static ColumnPath buildColumnPath(String columnFamily) {
		ColumnPath columnPath = new ColumnPath();
		columnPath.setColumn_family(columnFamily);
		return columnPath;
	}

	/**
	 * 从ColumnFamily中选取所有行的Rowkey
	 * 
	 * @param keySpace
	 *            keySpace名
	 * @param columnFamily
	 *            列族名
	 * @return Cassandra结果集，可以用next()方法遍历，用getKey()方法获得Rowkey
	 * @throws SQLException
	 *             若SQL语句不合法
	 * @throws ClassNotFoundException
	 *             若找不到驱动类
	 */
	/**
	 * 
	 * @param keySpace parameter
	 * @param columnFamily parameter
	 * @throws Exception 异常
	 * @return
	 */
	public static List<KeySlice> selectAll(final String keySpace,
			final String columnFamily) throws Exception {
		List<KeySlice> result;
		Cassandra.Client client = getInstance(keySpace);

		client.set_keyspace(keySpace);
		ColumnParent columnParent = new ColumnParent(columnFamily);

		SlicePredicate slicePredicate = buildSlicePredicate("", "");
		KeyRange keyrange = buildKeyRange("", "");

		result = client.get_range_slices(columnParent, slicePredicate,
				keyrange, ConsistencyLevel.ONE);

		return result;
	}

	/**
	 * 从ColumnFamily中选取所有行的Rowkey
	 * 
	 * @param keySpace
	 *            keySpace名
	 * @param columnFamily
	 *            列族名
	 * @return Cassandra结果集，可以用next()方法遍历，用getKey()方法获得Rowkey
	 * @throws SQLException
	 *             若SQL语句不合法
	 * @throws ClassNotFoundException
	 *             若找不到驱动类
	 */
	/**
	 * 
	 * @param keySpace parameter
	 * @param columnFamily parameter
	 * @param columnName parameter
	 * @throws Exception 异常
	 * @return
	 */
	public static List<KeySlice> selectAllInOneColumn(final String keySpace,
			final String columnFamily, String columnName) throws Exception {
		List<KeySlice> result;
		Cassandra.Client client = getInstance(keySpace);

		client.set_keyspace(keySpace);
		ColumnParent columnParent = new ColumnParent(columnFamily);

		SlicePredicate slicePredicate = buildSlicePredicate(columnName,
				columnName);
		KeyRange keyrange = buildKeyRange("", "");

		result = client.get_range_slices(columnParent, slicePredicate,
				keyrange, ConsistencyLevel.ONE);

		return result;
	}

	/**
	 * 从ColumnFamily中选取所有行的Rowkey
	 * 
	 * @param keySpace
	 *            keySpace名
	 * @param columnFamily
	 *            列族名
	 * @return Cassandra结果集，可以用next()方法遍历，用getKey()方法获得Rowkey
	 * @throws SQLException
	 *             若SQL语句不合法
	 * @throws ClassNotFoundException
	 *             若找不到驱动类
	 */
	/**
	 * 
	 * @param keySpace parameter
	 * @param columnFamily parameter
	 * @param columnName1 parameter
	 * @param value parameter
	 * @param columnName2 parameter
	 * @throws Exception 异常
	 * @return
	 */
	public static List<KeySlice> selectAllInOneColumnByAnother(
			final String keySpace, final String columnFamily,
			String columnName1, String value, String columnName2)
			throws Exception {
		List<KeySlice> result;
		Cassandra.Client client = getInstance(keySpace);

		client.set_keyspace(keySpace);
		ColumnParent columnParent = new ColumnParent(columnFamily);

		if (columnName1.toLowerCase().equals("key")) {
			SlicePredicate slicePredicate = buildSlicePredicate("", "");
			KeyRange keyrange = buildKeyRange(value, value);

			result = client.get_range_slices(columnParent, slicePredicate,
					keyrange, ConsistencyLevel.ONE);
		} else {
			IndexClause indexClause = buildIndexClause(columnName1, value,
					IndexOperator.EQ, "");
			SlicePredicate slicePredicate = buildSlicePredicate(columnName2,
					columnName2);

			result = client.get_indexed_slices(columnParent, indexClause,
					slicePredicate, ConsistencyLevel.ONE);
		}
		return result;
	}

	/**
	 * 从ColumnFamily中选取指定列的内容为指定值的Rowkey
	 * 
	 * @param keySpace
	 *            keySpace名
	 * @param columnFamily
	 *            列族名
	 * @param column
	 *            指定列的列名
	 * @param value
	 *            指定列的值
	 * @return Cassandra结果集，可以用next()方法遍历，用getKey()方法获得Rowkey
	 * @throws SQLException
	 */
	/**
	 * 
	 * @param keySpace parameter
	 * @param columnFamily parameter
	 * @param column parameter
	 * @param value parameter
	 * @throws Exception 异常
	 * @return
	 */
	public static List<KeySlice> select(final String keySpace,
			final String columnFamily, final String column, final String value)
			throws Exception {
		List<KeySlice> result;
		Cassandra.Client client = getInstance(keySpace);

		client.set_keyspace(keySpace);
		ColumnParent columnParent = new ColumnParent(columnFamily);

		if (column.toLowerCase().equals("key")) {
			SlicePredicate slicePredicate = buildSlicePredicate("", "");
			KeyRange keyrange = buildKeyRange(value, value);

			result = client.get_range_slices(columnParent, slicePredicate,
					keyrange, ConsistencyLevel.ONE);
		} else {
			IndexClause indexClause = buildIndexClause(column, value,
					IndexOperator.EQ, "");
			SlicePredicate slicePredicate = buildSlicePredicate("", "");

			result = client.get_indexed_slices(columnParent, indexClause,
					slicePredicate, ConsistencyLevel.ONE);
		}
		return result;
	}

	/**
	 * 从ColumnFamily中选取指定列的内容为指定值的Rowkey
	 * 
	 * @param keySpace
	 *            keySpace名
	 * @param columnFamily
	 *            列族名
	 * @param column
	 *            指定列的列名
	 * @param value
	 *            指定列的值
	 * @return Cassandra结果集，可以用next()方法遍历，用getKey()方法获得Rowkey
	 * @throws SQLException
	 */
	/**
	 * 
	 * @param keySpace parameter
	 * @param columnFamily parameter
	 * @param column parameter
	 * @param value parameter
	 * @param column1 parameter
	 * @param value1 parameter
	 * @throws Exception 异常
	 * @return
	 */
	public static List<KeySlice> selectByTwoColumn(final String keySpace,
			final String columnFamily, final String column, final String value,
			final String column1, final String value1) throws Exception {
		List<KeySlice> result;
		Cassandra.Client client = getInstance(keySpace);

		client.set_keyspace(keySpace);
		ColumnParent columnParent = new ColumnParent(columnFamily);

		if (column.toLowerCase().equals("key")) {
			SlicePredicate slicePredicate = buildSlicePredicate("", "");
			KeyRange keyrange = buildKeyRange(value, value);

			result = client.get_range_slices(columnParent, slicePredicate,
					keyrange, ConsistencyLevel.ONE);
		} else {
			IndexClause indexClause = buildIndexClause(column, value, column1,
					value1, IndexOperator.EQ, "");
			SlicePredicate slicePredicate = buildSlicePredicate("", "");

			result = client.get_indexed_slices(columnParent, indexClause,
					slicePredicate, ConsistencyLevel.ONE);
		}
		return result;
	}

	/**
	 * 更新数据库中某行某列的信息
	 * 
	 * @param keySpace
	 *            keySpace名
	 * @param columnFamily
	 *            列族名
	 * @param rowKey
	 *            对应行的行键
	 * @param column
	 *            目标列的列名
	 * @param value
	 *            目标列的值
	 * @throws SQLException
	 */
	/**
	 * 
	 * @param keySpace parameter
	 * @param columnFamily parameter
	 * @param columnName parameter
	 * @param rowKey parameter
	 * @param value parameter
	 * @throws Exception 异常
	 * @return
	 */
	public static void update(final String keySpace, final String columnFamily,
			final String rowKey, final String columnName, final String value)
			throws Exception {
		Cassandra.Client client = getInstance(keySpace);

		client.set_keyspace(keySpace);
		ColumnParent columnParent = new ColumnParent(columnFamily);

		Column column = buildColumn(columnName, value);

		client.insert(ByteBuffer.wrap(rowKey.getBytes()), columnParent, column,
				ConsistencyLevel.ONE);
	}

	/**
	 * 
	 * @param keySpace parameter
	 * @param columnFamily parameter
	 * @param columnName parameter
	 * @param rowKey parameter
	 * @param value parameter
	 * @throws Exception 异常
	 * @return
	 */
	public static void delete(final String keySpace, final String columnFamily,
			final String rowKey, final String columnName, final String value)
			throws Exception {
		Cassandra.Client client = getInstance(keySpace);

		client.set_keyspace(keySpace);
		ColumnParent columnParent = new ColumnParent(columnFamily);

		ColumnPath columnPath = buildColumnPath(columnFamily);
		Column column = buildColumn(columnName, value);
		client.remove(ByteBuffer.wrap(rowKey.getBytes()), columnPath,
				System.currentTimeMillis(), ConsistencyLevel.ONE);
		// List<KeySlice> result = select(keySpace, columnFamily, columnName,
		// value);
		// for (KeySlice ks : result) {
		// client.remove(ByteBuffer.wrap(rowKey.getBytes())), columnPath,
		// System.currentTimeMillis(), ConsistencyLevel.ALL);
		// }
	}

	/**
	 * 删除指定column列的值为value的行的信息
	 * 
	 * @param keySpace
	 *            keySpace名
	 * @param columnFamily
	 *            列族名
	 * @param column
	 *            指定列的列名
	 * @param value
	 *            指定列的value名
	 * @throws SQLException
	 */
	/**
	 * 
	 * @param keySpace parameter
	 * @param columnFamily parameter
	 * @param column parameter
	 * @param value parameter
	 * @throws Exception 异常
	 * @return
	 */
	public static void delete(final String keySpace, final String columnFamily,
			final String column, final String value) throws Exception {
		Cassandra.Client client = getInstance(keySpace);

		client.set_keyspace(keySpace);
		// ColumnParent columnParent = new ColumnParent(columnFamily);

		ColumnPath columnPath = buildColumnPath(columnFamily);
		List<KeySlice> result = select(keySpace, columnFamily, column, value);
		for (KeySlice ks : result) {
			client.remove(ByteBuffer.wrap(ks.getKey()), columnPath,
					System.currentTimeMillis(), ConsistencyLevel.ONE);
		}
	}
}
